package com.flowable.project.rocketmq;

import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.flowable.engine.RuntimeService;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.stereotype.Component;

import java.io.UnsupportedEncodingException;
import java.util.List;

@Component
public class PayConsumer {

    @Autowired
    private RuntimeService runtimeService;

    private DefaultMQPushConsumer consumer;

    private String consumerGroup = "flowable_consumer_group";

    public PayConsumer() throws MQClientException {
        consumer = new DefaultMQPushConsumer(consumerGroup);
        consumer.setNamesrvAddr(JmsConfig.NAME_SERVER);
        //默认是集群方式，可以改为广播，但是广播方式消费消息，这种方式当消费失败时不会进行消息的重发
//        consumer.setMessageModel(MessageModel.BROADCASTING);

        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_LAST_OFFSET) ;
//        主题以及主题下的那些标签

        //SQL92方式订阅消息,amount属性值在发送消息的时候添加
//        consumer.subscribe(JmsConfig.TOPIC, MessageSelector.bySql("amout>5"));
        consumer.subscribe(JmsConfig.TOPIC, "*");
        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {
                MessageExt msg = msgs.get(0);
                String topic = msg.getTopic();
                String body = null;
                try {
                    body = new String(msg.getBody(), "utf-8");
                } catch (UnsupportedEncodingException e) {
                    e.printStackTrace();
                }
                String tags = msg.getTags();
                String keys = msg.getKeys();
                runtimeService.messageEventReceived("isok",keys);
                System.out.println("topic=" + topic + ", tags=" + tags + ", keys=" + keys + ", msg="+ body+"的消息从消息队列发送成功！！！");
                return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;
            }
        });
        consumer.start();
        System.out.println("consumer start...");
    }
}
